package io.jnet.server.serverchannelcontext;

import java.io.IOException;
import java.nio.channels.AsynchronousSocketChannel;
import com.jfireframework.baseutil.collection.buffer.ByteBuf;
import com.jfireframework.baseutil.collection.buffer.DirectByteBuf;
import com.jfireframework.baseutil.concurrent.MPMCQueue;
import com.jfireframework.baseutil.resource.ResourceCloseAgent;
import com.jfireframework.baseutil.resource.ResourceCloseCallback;
import io.jnet.common.bufstorage.BufStorage;
import io.jnet.common.bufstorage.BufStorage.StorageType;
import io.jnet.common.decodec.FrameDecodec;
import io.jnet.common.streamprocessor.StreamProcessor;
import io.jnet.server.api.ReadHandler;
import io.jnet.server.api.ServerChannelContext;
import io.jnet.server.api.ServerListener;
import io.jnet.server.api.WriteHandler;
import io.jnet.server.read.ConcentrationReadHandler;
import io.jnet.server.util.ProcessorTask;
import io.jnet.server.write.SeparationWriteHandler;

public class ConcentrationServerChannelContext implements ServerChannelContext
{
    
    private final ReadHandler                              readHandler;
    private final WriteHandler                             writeHandler;
    private final ByteBuf<?>                               ioBuf;
    private final StreamProcessor[]                        processors;
    private final AsynchronousSocketChannel                socketChannel;
    private final MPMCQueue<ProcessorTask>                 tasks;
    private final ResourceCloseAgent<ServerChannelContext> closeAgent = new ResourceCloseAgent<ServerChannelContext>(this, new ResourceCloseCallback<ServerChannelContext>() {
                                                                          
                                                                          @Override
                                                                          public void onClose(ServerChannelContext resource)
                                                                          {
                                                                              try
                                                                              {
                                                                                  socketChannel.close();
                                                                              }
                                                                              catch (IOException e)
                                                                              {
                                                                                  e.printStackTrace();
                                                                              }
                                                                              finally
                                                                              {
                                                                                  ioBuf.release();
                                                                              }
                                                                          }
                                                                      });
    
    public ConcentrationServerChannelContext(//
            MPMCQueue<ProcessorTask> tasks, //
            BufStorage bufStorage, //
            int maxMerge, //
            ServerListener serverListener, //
            StreamProcessor[] processors, //
            AsynchronousSocketChannel socketChannel, //
            FrameDecodec frameDecodec)
    {
        if (bufStorage.type() == StorageType.array)
        {
            throw new UnsupportedOperationException("只支持队列类型的BufStorage");
        }
        this.tasks = tasks;
        this.socketChannel = socketChannel;
        this.processors = processors;
        ioBuf = DirectByteBuf.allocate(128);
        readHandler = new ConcentrationReadHandler(tasks, socketChannel, frameDecodec, ioBuf, serverListener, this);
        writeHandler = new SeparationWriteHandler(maxMerge, socketChannel, serverListener, bufStorage, this);
    }
    
    @Override
    public void registerRead()
    {
        readHandler.registerRead();
    }
    
    @Override
    public void commitWrite()
    {
        writeHandler.commitWrite();
    }
    
    @Override
    public boolean close()
    {
        return closeAgent.close();
    }
    
    @Override
    public boolean isOpen()
    {
        return socketChannel.isOpen();
    }
    
    @Override
    public void push(Object send, int index) throws Throwable
    {
        tasks.offerAndSignal(new ProcessorTask(send, index, this));
    }
    
    @Override
    public StreamProcessor[] processors()
    {
        return processors;
    }
    
}
